package dse;

import java.util.Collection;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Map;
import java.util.Vector;
import java.io.Serializable;

public class Cluster implements Serializable
{
	private static final long serialVersionUID = 1L;
	private static final int cacheTimeOut = 1440;
	
	public static final int InvalidId = -1;
	
	private int clusterId;
	private Node superNode;
	private Vector<Node> members;
	
	private LinkedList<DataObject> hostedDataObjects;
	private Hashtable<DataObject, Integer> cachedDataObjects; 
	private LinkedList<DataObject> unactivatedCachedDataObjects;
	private Vector<Cluster> neighborhood;
	private BloomFilter<DataObject> cachedDataObjectBloomFilter;
	private Map<Cluster, BloomFilter<DataObject>> neighborhoodBloomFilters;
	
	
	
	public Cluster (int id) {
		clusterId = id;
		members = new Vector<Node>();
		hostedDataObjects = new LinkedList<DataObject>();
		cachedDataObjects = new Hashtable<DataObject, Integer>();
		unactivatedCachedDataObjects = new LinkedList<DataObject>();
		neighborhood = new Vector<Cluster>();
		neighborhoodBloomFilters = 
			new Hashtable<Cluster, BloomFilter<DataObject>>();
	}
	
	public Cluster (int id, Collection<? extends Node> nodes) {
		this(id);
		assert nodes.size() > 0;
		
		for (Node n : nodes)
			addNode(n);
	}
	
	
	
	
	public int id () { return clusterId; }
	public void setId (int newId) {
		clusterId = newId;
		for (Node n : members)
			n.setClusterId(newId);
	} 
	
	public Node superNode () { return superNode; }
	public void setSuperNode (Node superNode) {
		if (this.superNode != null)
			this.superNode.setIsSupernode(false);
		this.superNode = superNode;
		this.superNode.setIsSupernode(true);
	}
	
	public void addNode (Node n) {
		if (members.size() == 0) {
			n.setIsSupernode(true);
			superNode = n;
		}
		n.setLocalId(members.size()+1);
		n.setClusterId(clusterId);
		members.add(n);
	}
	
	public Iterable<Node> nodes () {
		return members;
	}
	
	public int size () { return members.size(); }
	
	public Node lookupNode (int localNodeId) {
		return members.elementAt(localNodeId - 1);
	}
	
	/**
	 * This returns the maximum local id of nodes in this cluster.  This can be
	 * used in hashing.
	 *  
	 * @return
	 */
	public int maxLocalNodeId () {
		return members.size();
	}
	
	
	
	
	/**
	 * This notes that the data object is hosted in the cluster, but does not
	 * actually do anything to ensure that the data object or its shares are 
	 * really hosted here.
	 * 
	 * @param data
	 */
	public void host (DataObject data) {
		hostedDataObjects.add(data);
	}
	
	public boolean isHosting (DataObject data) {
		return hostedDataObjects.contains(data);
	}
	
	/**
	 * Adds a cached data object to the cluster.
	 * 
	 * @param data
	 * @param activate	Whether or not the cached object is immediatly active.
	 * 					If this is false, a later call to 
	 * 					activateCachedDataObject must be made before queries 
	 * 					for the cached object will succeed.
	 */
	public void addCachedDataObject (DataObject data, int curPeriod, boolean activate) {
		if (activate) {
			cachedDataObjects.put(data, curPeriod);
			cachedDataObjectBloomFilter.add(data);
		} else
			unactivatedCachedDataObjects.add(data);
	}
	
	public boolean isCaching (DataObject data, int curPeriod) {
		if(cachedDataObjects.get(data)==null)
			return false;
		else if((cachedDataObjects.get(data)+cacheTimeOut)<curPeriod)
		{
			//System.out.println("cached data object time out:" + cachedDataObjects.get(data) + ":" + curPeriod);
			
			cachedDataObjects.remove(data);
//			for(Node node : members)
//			{
//				node.removeCachedDataObject(data);
//			}
			return false;
		}
		else
			return true;
	}
	
	
	public Iterable<DataObject> cachedDataObjects () {
		return cachedDataObjects.keySet();
	}
	
	public boolean isCachingUnactive (DataObject data) {
		return unactivatedCachedDataObjects.contains(data);
	}
	
	public void activateCachedDataObject (DataObject data, int curPeriod) {
		if (unactivatedCachedDataObjects.contains(data) == false)
			return;
		unactivatedCachedDataObjects.remove(data);
		cachedDataObjects.put(data, curPeriod);
		cachedDataObjectBloomFilter.add(data);
	}
	
	public void removeUnactivatedCachedDataObject (DataObject data) {
		unactivatedCachedDataObjects.remove(data);
	}
	
	/**
	 * This may lead to unexpected behavior if not used with care.
	 * 
	 * @param data
	 */
	public void removeCachedDataObject (DataObject data) {
		cachedDataObjects.remove(data);
	}
	
	
	
	
	public void addNeighboringCluster (Cluster c) {
		neighborhood.add(c);
	}
	
	public Iterable<Cluster> neighborhood () {
		return neighborhood;
	}
	
	public int neighborhoodSize () {
		return neighborhood.size();
	}
	
	public BloomFilter<DataObject> bloomFilter () { 
		return cachedDataObjectBloomFilter;
	}
	
	public void setBloomFilter (BloomFilter<DataObject> bloomFilter) {
		this.cachedDataObjectBloomFilter = bloomFilter;
	}
	
	public Map<Cluster, BloomFilter<DataObject>> neighborhoodBloomFilters () {
		return neighborhoodBloomFilters;
	}
	
	public void refreshBloomFilter (
			Cluster cluster, 
			BloomFilter<DataObject> filter)
	{
		neighborhoodBloomFilters.put(cluster, filter);
	}
	
	
	
	/**
	 * This method should be overriden if possible to provide a better distance
	 * calculation between clusters.  This method need not be exact but since 
	 * it is just used to order clusters by their distance to another cluster.
	 * 
	 * @param other
	 * @return
	 */
	public double distanceTo (Cluster other) {
		return 0;
	}
	
	
		
	/**
	 * Useful for debugging.
	 */
	@Override
	public String toString () {
		String res = String.format(
				"cluster#%d{size=%d}", 
				new Object [] {clusterId,members.size()}
				);
		return res;
	}
}
